package org.apache.flink.formats.jsontm;

import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;

import java.io.Serializable;
import java.util.*;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * 将字段路径进行管理, 组织成树
 * @author guan
 */
public class JsonTmPathManager implements Serializable {
    private final String[] paths;
    private final Tree root;

    /**
     * @param paths 字段路径集, .或者/分割
     */
    public JsonTmPathManager(String[] paths) {
        this.paths = paths;

        List<String[]> pathArrays = Stream.of(paths).map(item -> item.split("\\.|/")).collect(Collectors.toList());
        this.root = new Tree();

        //转成树的结构, 树我字段的路径, 节点值为字段下标索引 1:a.b =  a-> b->1
        Tree parent;
        for (int fieldI = 0; fieldI < pathArrays.size(); fieldI++) {
            String[] fieldPath = pathArrays.get(fieldI);
            parent = root;
            for (int i = 0; i < fieldPath.length-1; i++) {
                parent = parent.getChild(fieldPath[i]);
            }
            parent.addNode(fieldPath[fieldPath.length-1], fieldI);
        }
    }

    public String[] getPaths() {
        return paths;
    }

    public Tree getRoot() {
        return root;
    }

    @Override
    public String toString() {
        return Arrays.toString(paths);
    }

    /**
     * 树的节点
     */
    public static class Node implements Serializable {
        /**
         * 节点深度
         */
        private final int dept;
        /**
         * 父节点, 根节点此处为空
         */
        private final Tree parent;
        /**
         * 当前节点路径 dept = path.split("\\.|/").length
         */
        private final String path;
        /**
         * 当前节点存的值, 只有叶子节点有值, 路径节点此处为空
         */
        private final Set<Integer> fields = new TreeSet<>();

        private Node(Tree parent, String path, int dept) {
            this.parent = parent;
            this.path = path;
            this.dept = dept;
        }

        public int getDept() {
            return dept;
        }

        public Tree getParent() {
            return parent;
        }

        public Set<Integer> getFields() {
            return fields;
        }

        public void addFields(int field) {
            fields.add(field);
        }

        public String getPath() {
            return path;
        }

        public String getFullPath() {
            if (this.parent == null) {
                return this.path;
            }
            String fullPath = parent.getFullPath();
            if (fullPath.isEmpty()) {
                return this.path;
            }
            return this.parent.getFullPath() + "." + this.path;
        }

        public List<Node> getNodes() {
            return Collections.emptyList();
        }

        public List<Tree> getChilds() {
            return Collections.emptyList();
        }

        /**
         * 消费数据
         * @param cache 缓存其他字段
         * @param parentJsonValue 父节点节点值
         * @param out 消费者
         */
        void consumer(JsonNode[] cache, JsonNode parentJsonValue, Consumer<JsonNode[]> out) {
            if (parentJsonValue != null && parentJsonValue.isArray()) {
                for (JsonNode valueItem : parentJsonValue) {
                    consumer(cache, valueItem, out);
                }
            } else {
                String path = getPath();
                // 值可以为空
                JsonNode jsonValue = parentJsonValue == null || path.isEmpty() ? parentJsonValue : parentJsonValue.get(path);
                if (jsonValue != null && jsonValue.isArray()) {
                    for (JsonNode node : jsonValue) {
                        // 每一个值, 都需要消费一次
                        consumerChildAndNode(cache, node, out);
                    }
                } else {
                    consumerChildAndNode(cache, jsonValue, out);
                }
            }
        }

        /**
         * 消费子节点和数据节点
         * @param cache 缓存其他字段
         * @param currJsonValue 当前节点值
         * @param out 消费者
         */
        void consumerChildAndNode(JsonNode[] cache, JsonNode currJsonValue, Consumer<JsonNode[]> out) {
            for (Node node : getNodes()) {
                // 值可以为空
                JsonNode nodeValue = currJsonValue == null ? null : currJsonValue.get(node.getPath());
                if (nodeValue != null && nodeValue.isArray()) {
                    // 这里可以判断, 当前node类似是ARRAY时, 在这里特殊处理, 暂时不支持吧
                    Consumer<JsonNode[]> finalOut = out;
                    out = jsonNodes -> {
                        for (JsonNode nodeValueItem : nodeValue) {
                            for (Integer field : node.getFields()) {
                                jsonNodes[field] = nodeValueItem;
                            }
                            // 每一个值, 都需要消费一次
                            finalOut.accept(jsonNodes);
                        }
                    };
                } else {
                    for (Integer field : node.getFields()) {
                        cache[field] = nodeValue;
                    }
                }
            }
            List<Tree> childs = getChilds();
            if (childs.size() == 1) {
                // 单个的特殊处理一下, 不用多套一层消费函数了
                childs.get(0).consumer(cache, currJsonValue, out);
            } else {
                for (Tree child : childs) {
                    Consumer<JsonNode[]> finalOut = out;
                    out = jsonNodes -> child.consumer(jsonNodes, currJsonValue, finalOut);
                }
                out.accept(cache);
            }
        }

        @Override
        public String toString() {
            return getFullPath();
        }
    }

    /**
     * 树的节点
     */
    public static class Tree extends Node implements Serializable {
        /**
         * 叶子节点
         */
        private final List<Node> nodes = new ArrayList<>();
        /**
         * 路径节点
         */
        private final List<Tree> childs = new ArrayList<>();

        public Tree() {
            this(null, "", 0);
        }

        private Tree(Tree parent, String path, int dept) {
            super(parent, path, dept);
        }

        @Override
        public List<Node> getNodes() {
            return nodes;
        }

        @Override
        public List<Tree> getChilds() {
            return childs;
        }

        public void addNode(String path, int index) {
            this.getNode(path).addFields(index);
        }

        /**
         * 获取路径对应树节点追加使用, 如果不存在创建新的
         * @param path 路径
         */
        private Tree getChild(String path) {
            for (Tree tree : childs) {
                if (tree.getPath().equals(path)) {
                    return tree;
                }
            }
            Tree child = new Tree(this, path, getDept() + 1);
            childs.add(child);
            return child;
        }

        /**
         * 获取路径对应节点追加使用, 如果不存在创建新的
         * @param path 路径
         */
        private Node getNode(String path) {
            for (Node node : nodes) {
                if (node.getPath().equals(path)) {
                    return node;
                }
            }
            Node node = new Node(this, path, getDept() + 1);
            nodes.add(node);
            return node;
        }


        @Override
        public String toString() {
            return getFullPath();
        }
    }


    /**
     * 消费数据
     * @param jsonNode 数据
     * @param out 消费者
     */
    public void consumer(JsonNode jsonNode, Consumer<JsonNode[]> out) {
        root.consumer(new JsonNode[paths.length], jsonNode, values -> {
            for (JsonNode value : values) {
                if (value != null) {
                    // 过滤所有字段均为空的数据
                    out.accept(values);
                    break;
                }
            }
        });
    }
}
